package platon

import (
	"context"
	"fmt"
	"platon-tools/go-service/common/constant"
	"platon-tools/go-service/common/util"
	"platon-tools/go-service/event"
	go_logger "platon-tools/go-service/go-logger"
	"platon-tools/go-service/module"
	"platon-tools/go-service/module/model"
	"platon-tools/go-service/protocol/rpc_pb"
	"platon-tools/go-service/service"
	"sync"
	"sync/atomic"
	"time"
)

type Platon struct {
	sync.WaitGroup
	ctx    context.Context
	cancel context.CancelFunc

	addr  string
	topic string

	service     *service.Service
	latestBlock uint64

	running int32
}

func (p *Platon) Start(ctx context.Context, config *service.Config, s *service.Service) error {

	fmt.Println("platon monitor init start")
	p.addr = fmt.Sprintf("%s:%s", config.Platon.KafkaIP, config.Platon.KafkaPort)
	p.topic = config.Platon.KafkaTopic
	p.service = s
	p.ctx = ctx

	atomic.StoreInt32(&p.running, 1)

	go p.runLoop()

	fmt.Println("platon monitor init success")

	return nil
}

func (p *Platon) runLoop() {
	defer p.Close()

	ch := make(chan event.Msg, 5)
	defer close(ch)

	sub := p.service.GetEvent().Subscribe(ch)

	defer sub.Unsubscribe()
	go func() {
		for {
			select {
			case <-p.ctx.Done():
				return
			case <-sub.Err():
				return
			case msg, ok := <-ch:
				if ok {
					switch msg.Type {
					case rpc_pb.MsgType_NodeMonitorItem:
						if monitor, ok := msg.Payload.(event.NodeMonitorItem); ok {
							go_logger.Logger.Sugar().Debugf("订阅消息 monitor item【%+v】", monitor.NodeId)
							err := p.MonitorListTrans(monitor)
							if err != nil {
								go_logger.Logger.Error(fmt.Sprintf("validator message save error: %s", err.Error()))
							}
						}
					case rpc_pb.MsgType_Validator:
						if vlist, ok := msg.Payload.(event.ValidatorList); ok {
							go_logger.Logger.Sugar().Debugf("订阅消息 共识轮【%d】", vlist.ConsensusNo)
							go_logger.Logger.Sugar().Debugf("出块奖励%v,委托总量%v", vlist.MiningReward, vlist.NodesList[0].DelegationSum)
							err := p.ContractConsensusTrans(vlist)
							if err != nil {
								go_logger.Logger.Error(fmt.Sprintf("vlidator message save error: %s", err.Error()))
							}
						}
					case rpc_pb.MsgType_Verifier:
						if vlist, ok := msg.Payload.(event.VerifierList); ok {
							go_logger.Logger.Sugar().Debugf("订阅消息 结算轮【%d】", vlist.Epoch)
							err := p.ContractEpochTrans(vlist)
							if err != nil {
								go_logger.Logger.Error(fmt.Sprintf("virifier message save error: %s", err.Error()))
							}
						}
					}
				}
			}
		}
	}()

	sugar := go_logger.Logger.Sugar()

	ticker := time.NewTicker(time.Minute * 1)
	for {
		select {
		case <-ticker.C:
			go p.CheckMonitor()
		case <-p.ctx.Done():
			sugar.Infof("platon monitor event loop quit timer: %s", time.Now().String())
			return
		}
	}
}

func (p *Platon) IsRunning() bool {
	if atomic.LoadInt32(&p.running) == 1 {
		return true
	}

	return false
}

func (p *Platon) Name() string {
	return "PLATON"
}

func (p *Platon) Close() error {
	if p.IsRunning() {
		atomic.StoreInt32(&p.running, 0)
		p.cancel()
	}

	return nil
}

// 处理传过来的共识轮数据
func (p *Platon) ContractConsensusTrans(nodeList event.ValidatorList) error {

	var err error

	if len(nodeList.NodesList) == 0 {
		go_logger.Logger.Warn(fmt.Sprintf("no node contract item data"))
		return nil
	}

	r := p.service.ModuleByName("MYSQLDB")
	if db, ok := r.(*module.MysqlDB); ok {

		// 获取所有需要监控的节点
		var platNodes []*model.SysNode
		platNodesMap := make(map[string]*model.SysNode)
		err = db.Table(model.SysNodeTable).Find(&platNodes).Error
		if err != nil {
			go_logger.Logger.Error(err.Error())
			return err
		}
		// 需要监控的节点map
		for _, sysNode := range platNodes {
			platNodesMap[sysNode.NodeId] = sysNode
		}

		for _, nodeItem := range nodeList.NodesList {
			sysNode, ok := platNodesMap[nodeItem.NodeId]

			//记录历史
			nodeContractHistory := model.SysNodeContractHistory{
				HistoryType:          model.HISTORY_TYPE_CONSENSUS,
				CurrentBlock:         nodeList.CurrentBlock,
				Epoch:                nodeList.Epoch,
				ConsensusNo:          nodeList.ConsensusNo,
				NodeId:               nodeItem.NodeId,
				NodeName:             nodeItem.NodeName,
				StakingAddr:          nodeItem.StakingAddr,
				StakingAddrAccount:   nodeItem.StakingAddrAccount,
				StakingReward:        0,
				MiningReward:         nodeList.MiningReward,
				DelegationReward:     0,
				BenefitAddr:          nodeItem.BenefitAddr,
				BenefitAddrAccount:   nodeItem.BenefitAddrAccount,
				BonusRatio:           nodeItem.BonusRatio,
				DelegatorSum:         0,
				StakingSum:           nodeItem.StakingSum,
				DelegationSum:        nodeItem.DelegationSum,
				BenefitSum:           0,
				DelegationNotReceive: 0,
				CurrentBlockingNode:  nodeList.CurrentBlockNode,
				Status:               nodeItem.NodeStatus,
				NextPeriod:           10750 * (nodeList.CurrentBlock/10750 + 1),
				CreateTime:           time.Now(),
				UpdateTime:           time.Now(),
			}

			if !ok {
				err = db.Table(model.SysNodeContractHistoryTable).Save(&nodeContractHistory).Error
				if err != nil {
					go_logger.Logger.Error(err.Error())
					return err
				}
				continue
			}

			var nodeContract model.SysNodeContract
			err = db.Table(model.SysNodeContractTable).Where("node_id = ?", sysNode.NodeId).First(&nodeContract).Error
			if err != nil {
				go_logger.Logger.Error(err.Error())
				continue
			}

			// 块高没变，则不更新
			if nodeContract.CurrentBlock >= nodeList.CurrentBlock {
				go_logger.Logger.Warn(fmt.Sprintf("块高导致未更新，当前块高: %d, 推送块高: %d", nodeContract.CurrentBlock, nodeList.CurrentBlock))
				continue
			}

			// 共识周期轮次没变，则不更新
			if nodeContract.ConsensusNo >= nodeList.ConsensusNo {
				go_logger.Logger.Warn(fmt.Sprintf("共识轮次导致未更新，当前共识轮次: %d, 推送共识轮次: %d", nodeContract.ConsensusNo, nodeList.ConsensusNo))
				continue
			}

			nodeContractHistory.NodePrimaryId = sysNode.Id
			err = db.Table(model.SysNodeContractHistoryTable).Save(&nodeContractHistory).Error
			if err != nil {
				go_logger.Logger.Error(err.Error())
				continue
			}

			// 不关注的节点，只保存历史
			if !ok {
				continue
			}

			// 对于当前状态的修改
			nodeContract.CurrentBlock = nodeList.CurrentBlock
			nodeContract.Epoch = nodeList.Epoch
			nodeContract.ConsensusNo = nodeList.ConsensusNo
			nodeContract.BonusRatio = nodeItem.BonusRatio
			nodeContract.BenefitAddr = nodeItem.BenefitAddr
			nodeContract.BenefitAddrAccount = nodeItem.BenefitAddrAccount
			nodeContract.StakingAddr = nodeItem.StakingAddr
			nodeContract.StakingAddrAccount = nodeItem.StakingAddrAccount
			nodeContract.StakingSum = nodeItem.StakingSum
			nodeContract.DelegationSum = nodeItem.DelegationSum
			nodeContract.MiningReward += nodeList.MiningReward
			nodeContract.CurrentBlockingNode = nodeList.CurrentBlockNode
			nodeContract.TotalBalance = nodeItem.TotalBalance
			nodeContract.Status = nodeItem.NodeStatus
			nodeContract.UpdateTime = time.Now()
			nodeContract.NextPeriod = 10750 * (nodeList.CurrentBlock/10750 + 1)
			err = db.Table(model.SysNodeContractTable).Save(&nodeContract).Error
			if err != nil {
				go_logger.Logger.Error(err.Error())
				return err
			}

			stakingKey := fmt.Sprintf("staking_reward_%d", sysNode.Id)
			delegationKey := fmt.Sprintf("delegation_reward_%d", sysNode.Id)
			score := util.ToString(time.Now().Format(constant.DATE_DAY_VALUE_FORMAT))

			s := p.service.ModuleByName("REDIS")
			if redis, ok := s.(*module.RedisDB); ok {

				_, err = redis.ZAdd(delegationKey, fmt.Sprintf("%.2f", nodeContract.DelegationReward), score)
				if err != nil {
					go_logger.Logger.Error(delegationKey + " 存储失败, score: " + score)
					return err
				}

				_, err = redis.ZAdd(stakingKey, fmt.Sprintf("%.2f", nodeContract.StakingReward), score)
				if err != nil {
					go_logger.Logger.Error(stakingKey + " 存储失败, score: " + score)
					return err
				}
			}
		}

	}
	return nil
}

// 处理传过来的结算轮数据
func (p *Platon) ContractEpochTrans(nodeList event.VerifierList) error {

	var err error

	if len(nodeList.NodesList) == 0 {
		go_logger.Logger.Warn(fmt.Sprintf("no node contract item data"))
		return nil
	}

	r := p.service.ModuleByName("MYSQLDB")
	if db, ok := r.(*module.MysqlDB); ok {

		// 获取所有需要监控的节点
		var platNodes []*model.SysNode
		platNodesMap := make(map[string]*model.SysNode)
		err = db.Table(model.SysNodeTable).Find(&platNodes).Error
		if err != nil {
			go_logger.Logger.Error(err.Error())
			return err
		}
		// 需要监控的节点map
		for _, sysNode := range platNodes {
			platNodesMap[sysNode.NodeId] = sysNode
		}

		for _, nodeItem := range nodeList.NodesList {
			sysNode, ok := platNodesMap[nodeItem.NodeId]

			//记录历史
			nodeContractHistory := model.SysNodeContractHistory{
				HistoryType:          model.HISTORY_TYPE_EPOCH,
				CurrentBlock:         nodeList.CurrentBlock,
				Epoch:                nodeList.Epoch,
				ConsensusNo:          nodeList.ConsensusNo,
				NodeId:               nodeItem.NodeId,
				NodeName:             nodeItem.NodeName,
				StakingAddr:          nodeItem.StakingAddr,
				StakingAddrAccount:   nodeItem.StakingAddrAccount,
				StakingReward:        nodeList.StakingReward,
				MiningReward:         0,
				BenefitAddr:          nodeItem.BenefitAddr,
				BenefitAddrAccount:   nodeItem.BenefitAddrAccount,
				BonusRatio:           nodeItem.BonusRatio,
				DelegatorSum:         0,
				StakingSum:           nodeItem.StakingSum,
				DelegationSum:        nodeItem.DelegationSum,
				DelegationNotReceive: nodeItem.DelegationNotReceive,
				CurrentBlockingNode:  nodeList.CurrentBlockNode,
				Status:               nodeItem.NodeStatus,
				NextPeriod:           10750 * (nodeList.CurrentBlock/10750 + 1),
				CreateTime:           time.Now(),
				UpdateTime:           time.Now(),
			}

			if !ok {
				nodeContractHistory.DelegationReward = 0
				nodeContractHistory.BenefitSum = nodeList.StakingReward

				err = db.Table(model.SysNodeContractHistoryTable).Save(&nodeContractHistory).Error
				if err != nil {
					go_logger.Logger.Error(err.Error())
					return err
				}

			}

			var nodeContract model.SysNodeContract
			err = db.Table(model.SysNodeContractTable).Where("node_id = ?", sysNode.NodeId).First(&nodeContract).Error
			if err != nil {
				go_logger.Logger.Error(err.Error())
				return err
			}

			// 计算本结算周期所有出块奖励
			var miningReward float64
			err = db.Raw("select count(mining_reward) from node_contract_history where epoch = ?", nodeList.Epoch).Find(&miningReward).Error
			if err != nil {
				go_logger.Logger.Error(err.Error())
				return err
			}

			// 委托收益 = (出块收益 + 质押收益) * 分红比例
			delegationReward := (nodeContract.MiningReward + nodeList.StakingReward) * float64(nodeItem.BonusRatio) / 10000

			nodeContractHistory.NodePrimaryId = sysNode.Id
			nodeContractHistory.DelegationSum = delegationReward
			nodeContractHistory.BenefitSum = miningReward + nodeList.StakingReward

			err = db.Table(model.SysNodeContractHistoryTable).Save(&nodeContractHistory).Error
			if err != nil {
				go_logger.Logger.Error(err.Error())
				continue
			}

			// 块高没变，则不更新
			if nodeContract.CurrentBlock >= nodeList.CurrentBlock {
				go_logger.Logger.Warn(fmt.Sprintf("块高导致未更新，当前块高: %d, 推送块高: %d", nodeContract.CurrentBlock, nodeList.CurrentBlock))
				continue
			}

			// 共识周期轮次没变，则不更新
			if nodeContract.ConsensusNo >= nodeList.ConsensusNo {
				go_logger.Logger.Warn(fmt.Sprintf("共识轮次导致未更新，当前共识轮次: %d, 推送共识轮次: %d", nodeContract.ConsensusNo, nodeList.ConsensusNo))
				continue
			}

			nodeContract.CurrentBlock = nodeList.CurrentBlock
			nodeContract.Epoch = nodeList.Epoch
			nodeContract.ConsensusNo = nodeList.ConsensusNo
			nodeContract.StakingAddr = nodeItem.StakingAddr
			nodeContract.StakingAddrAccount = nodeItem.StakingAddrAccount
			nodeContract.BonusRatio = nodeItem.BonusRatio
			nodeContract.StakingSum = nodeItem.StakingSum
			nodeContract.DelegationSum = nodeItem.DelegationSum
			nodeContract.StakingReward += nodeList.StakingReward
			nodeContract.DelegationReward += delegationReward
			nodeContract.MiningReward += miningReward
			nodeContract.TotalReward = nodeContract.TotalReward + nodeList.StakingReward + miningReward
			nodeContract.TotalBalance = nodeContract.TotalBalance
			nodeContract.DelegationNotReceive = nodeItem.DelegationNotReceive
			nodeContract.CurrentBlockingNode = nodeList.CurrentBlockNode
			nodeContract.Status = nodeItem.NodeStatus
			nodeContract.NextPeriod = 10750 * (nodeList.CurrentBlock/10750 + 1)
			nodeContract.UpdateTime = time.Now()
			err = db.Table(model.SysNodeContractTable).Save(&nodeContract).Error
			if err != nil {
				go_logger.Logger.Error(err.Error())
				return err
			}

			stakingKey := fmt.Sprintf("staking_reward_%d", sysNode.Id)
			delegationKey := fmt.Sprintf("delegation_reward_%d", sysNode.Id)
			score := time.Now().Format(constant.DATE_DAY_VALUE_FORMAT)

			s := p.service.ModuleByName("REDIS")
			if redis, ok := s.(*module.RedisDB); ok {

				_, err = redis.ZAdd(delegationKey, fmt.Sprintf("%.2f", nodeContract.DelegationReward), score)
				if err != nil {
					go_logger.Logger.Error(delegationKey + " 存储失败, score: " + score)
					return err
				}

				_, err = redis.ZAdd(stakingKey, fmt.Sprintf("%.2f", nodeContract.StakingReward), score)
				if err != nil {
					go_logger.Logger.Error(stakingKey + " 存储失败, score: " + score)
					return err
				}
			}
		}

	}
	return nil
}

// 处理传过来的监控数据
func (p *Platon) MonitorListTrans(monitorItem event.NodeMonitorItem) error {

	var err error

	r := p.service.ModuleByName("MYSQLDB")
	if db, ok := r.(*module.MysqlDB); ok {

		// 获取所有需要监控的节点
		var platNodes []*model.SysNode
		platNodesMap := make(map[string]*model.SysNode)
		err = db.Table(model.SysNodeTable).Find(&platNodes).Error
		if err != nil {
			go_logger.Logger.Error(err.Error())
			return err
		}
		// 需要监控的节点map
		for _, sysNode := range platNodes {
			platNodesMap[sysNode.NodeId] = sysNode
		}

		sysNode, ok := platNodesMap[monitorItem.NodeId]

		//记录历史
		nodeMonitorHistory := model.SysNodeMonitorHistory{
			NodeId:           monitorItem.NodeId,
			NodeName:         monitorItem.NodeName,
			DiskRatio:        monitorItem.DiskRatio,
			CPURatio:         monitorItem.CPURatio,
			MemoryRatio:      monitorItem.MemoryRatio,
			NetWorkRatio:     monitorItem.NetWorkRatio,
			IORatio:          monitorItem.IORatio,
			RAmount:          monitorItem.RAmount,
			WAmount:          monitorItem.WAmount,
			RWAmount:         monitorItem.RWAmount,
			RPCResponseTime:  monitorItem.RPCResponseTime,
			NodeCurrentHigh:  monitorItem.CurrentBlock,
			TotalCurrentHigh: monitorItem.CurrentHigh,
			NodeStatus:       util.ConvertMonitorStatus(monitorItem.NodeStatus),
			CreateTime:       time.Now(),
		}
		if sysNode != nil {
			nodeMonitorHistory.NodePrimaryId = sysNode.Id
		}
		err = db.Table(model.SysNodeMonitorHistoryTable).Save(&nodeMonitorHistory).Error
		if err != nil {
			go_logger.Logger.Error(err.Error())
			return err
		}

		// 不关注的节点，只保存历史
		if !ok {
			return nil
		}

		var nodeMonitor model.SysNodeMonitor
		err = db.Table(model.SysNodeMonitorTable).Where("node_id = ?", monitorItem.NodeId).First(&nodeMonitor).Error
		if err != nil {
			go_logger.Logger.Error(err.Error())
			return err
		}

		// 块高没变，则不更新
		if nodeMonitor.NodeCurrentHigh >= monitorItem.CurrentBlock {
			return nil
		}

		nodeMonitor.NodeId = monitorItem.NodeId
		nodeMonitor.NodeName = monitorItem.NodeName
		nodeMonitor.DiskRatio = monitorItem.DiskRatio
		nodeMonitor.CPURatio = monitorItem.CPURatio
		nodeMonitor.MemoryRatio = monitorItem.MemoryRatio
		nodeMonitor.NetWorkRatio = monitorItem.NetWorkRatio
		nodeMonitor.IORatio = monitorItem.IORatio
		nodeMonitor.RAmount = monitorItem.RAmount
		nodeMonitor.WAmount = monitorItem.WAmount
		nodeMonitor.RWAmount = monitorItem.RWAmount
		nodeMonitor.RPCResponseTime = monitorItem.RPCResponseTime
		nodeMonitor.NodeCurrentHigh = monitorItem.CurrentBlock
		nodeMonitor.TotalCurrentHigh = monitorItem.CurrentHigh
		nodeMonitor.NodeStatus = util.ConvertMonitorStatus(monitorItem.NodeStatus)
		nodeMonitor.UpdateTime = time.Now()
		err = db.Table(model.SysNodeMonitorTable).Save(&nodeMonitor).Error
		if err != nil {
			go_logger.Logger.Error(err.Error())
			return err
		}

	}
	return nil
}
